/** **************************************************
 * Description: 同步的缓存
 * Copyright:   Copyright (c) 2008
 * Company:    cl
 * @author     caoxl
 * @version    1.0
 * @see
 HISTORY
 *  2009-1-22 caoxl Create File
 ************************************************* */
package com.bfsuol.bfsuolframework.core.synchronization;

import org.apache.log4j.Logger;
import org.jgroups.*;
import org.jgroups.util.Util;

import java.awt.event.ActionEvent;
import java.awt.event.ActionListener;

/**
 * public class SecuritySyncGroup extends SyncGroup {<br>
 *   public SecuritySyncGroup() throws Exception{<br>
 * 	    super("GROUP_NAME");<br>
 *   }<br>
 *   public void clear() {}<br>
 *   public void init() {}<br>
 *   public void put(Long id) {}<br>
 *   public void reload() {}<br>
 *   public void remove(Long id) {}<br>
 *   public void execute(params) {}<br>
 * }
 * @author 新龙
 *
 */
public abstract class SyncGroup extends ReceiverAdapter implements ActionListener, ChannelListener{

	protected final Logger logger = Logger.getLogger(getClass());
	
	protected String groupName = "DEFALUT_GROUP";
//	protected String props = "UDP(mcast_addr=224.0.0.35;mcast_port=45566;ip_ttl=32;" +
//		"mcast_send_buf_size=150000;mcast_recv_buf_size=80000):" +
//		"PING(timeout=2000;num_initial_members=3):" +
//		"MERGE2(min_interval=5000;max_interval=10000):" +
//		"FD_SOCK:" +
//		"VERIFY_SUSPECT(timeout=1500):" +
//		"pbcast.STABLE(desired_avg_gossip=20000):" +
//		"pbcast.NAKACK(gc_lag=50;retransmit_timeout=300,600,1200,2400,4800):" +
//		"UNICAST(timeout=5000;min_wait_time=2000):" +
//		"FRAG(frag_size=4096;down_thread=false;up_thread=false):" +
//		"pbcast.GMS(join_timeout=5000;join_retry_timeout=2000;" +
//		"shun=false;print_local_addr=true)";
	
//	protected String props = "TCP(start_port=7800):" +
//		"TCPPING(initial_hosts=localhost[7800];port_range=5;timeout=3000;"+
//		"num_initial_members=3;up_thread=true;down_thread=true):" +
//		"VERIFY_SUSPECT(timeout=1500;down_thread=false;up_thread=false):"+
//		"pbcast.STABLE(desired_avg_gossip=20000;down_thread=false;up_thread=true)"+
//		"pbcast.NAKACK(down_thread=true;up_thread=true;gc_lag=100;retransmit_timeout=300,600,1200,2400,4800"+
//		"pbcast.GMS(join_timeout=5000;join_retry_timeout=2000;shun=false;"+
//		"print_local_addr=false;down_thread=true;up_thread=true)";
		
	
//	protected String props = 
//		"TCP(start_port=7800):"+
//		"TCPPING(timeout=3000;initial_hosts=192.168.1.100[7800],192.168.1.199[7800];port_range=5;num_initial_members=3):"+
//        "MERGE2(max_interval=100000;min_interval=20000):"+
//        "FD(timeout=10000;max_tries=5;shun=true):"+
//        "VERIFY_SUSPECT(timeout=1500):"+
//        "pbcast.NAKACK(use_mcast_xmit=false;gc_lag=0;retransmit_timeout=300,600,1200,2400,4800;discard_delivered_msgs=true):"+
//        "UNICAST(timeout=300,600,1200):"+
//        "pbcast.STABLE(stability_delay=1000;desired_avg_gossip=50000;max_bytes=400000):"+
//        "VIEW_SYNC(avg_send_interval=60000):"+
//        "pbcast.GMS(print_local_addr=true;join_timeout=3000;shun=true;view_bundling=true):"+
//        "FC(max_credits=2000000;min_threshold=0.10):"+
//        "FRAG2(frag_size=60000)";
	
	private Channel channel = null;
	private int memberSize = 1;
	private boolean use_state = false;
	private long state_timeout = 60000;
	
	public SyncGroup(String groupName) throws Exception{
		this(groupName,true);
	}
	
	public SyncGroup(String groupName,boolean block) throws Exception{
		this.groupName = groupName;
		
		//channel = new JChannel(props);
		channel = new JChannel();
		//TODO Jgroup API 修改，这里需要重构
		/*
        if(block){
            channel.setOpt(Channel.BLOCK, Boolean.TRUE);
        }
        channel.setOpt(Channel.AUTO_RECONNECT, Boolean.TRUE);
        */
        channel.setReceiver(this);
        channel.addChannelListener(this);
        
        //初始化
        init();
        
        //连接
        if(!use_state) {
            channel.connect(groupName);
        }else{
        	channel.connect(groupName,null, state_timeout);
        }
	}
	
	/**
	 * 返回该缓存的名称，每个缓存的名称必须不同
	 * @return 缓存名称
	 */
	public String getGroupName() {
        return groupName;
    }

	/**
	 * @return the memberSize
	 */
	public int getMemberSize() {
		return memberSize;
	}

	protected boolean send(int action, long id, String params){
		try{
			SyncGroupCommand comm = new SyncGroupCommand(action, id, params);
			
			byte[] buf=Util.streamableToByteBuffer(comm);
            channel.send(new Message(null, null, buf));
            return true;
		}catch(Exception e){
			logger.error(e);
			return false;
		}
	}
	
	public void put(Long id){
		send(SyncGroupCommand.ACTION_PUT, id,null);
	}
	public void remove(Long id){
		send(SyncGroupCommand.ACTION_REMOVE, id,null);
	}
	public void reload(String params){
		send(SyncGroupCommand.ACTION_RELOAD, -1,params);
	}
	public void clear(String params){
		send(SyncGroupCommand.ACTION_CLEAR, -1,params);
	}
	public void custom(Long id, String params){
		send(SyncGroupCommand.ACTION_CUSTOM, id, params);
	}
	
	public void receive(Message msg) {
        if(logger.isInfoEnabled()){
        	logger.info(channel.getClusterName() + "  -- 接收到消息!");
        }
		byte[] buf=msg.getRawBuffer();
        if(buf == null) {
            return;
        }

        try {
        	SyncGroupCommand comm=(SyncGroupCommand)Util.streamableFromByteBuffer(SyncGroupCommand.class, buf, msg.getOffset(), msg.getLength());
        	switch(comm.getAction()) {
                case SyncGroupCommand.ACTION_PUT:
                	executePut(comm.getId());
                    break;
                case SyncGroupCommand.ACTION_REMOVE:
                	executeRemove(comm.getId());
                    break;
                case SyncGroupCommand.ACTION_RELOAD:
                	executeReload(comm.getParams());
                    break;
                case SyncGroupCommand.ACTION_CLEAR:
                	executeClear(comm.getParams());
                	break;
                case SyncGroupCommand.ACTION_CUSTOM:
                	execute(comm.getId(), comm.getParams());
                	break;
                default:
                	logger.error(channel.getClusterName() + " -- 接收到的动作未知 ： action" + comm.getAction());
                    break;
            }
            
        	if(logger.isInfoEnabled()){
        		logger.info(channel.getClusterName() + "  -- 处理消息! action="+comm.getAction() + ", id="+comm.getId());
        	}
            
        }catch(Exception e) {
            logger.error(e);
        }
    }

	protected void init(){};
	
	protected void executePut(Long id){};
	protected void executeRemove(Long id){};
	protected void executeReload(String params){};
	protected void executeClear(String params){};
	protected void execute(Long id, String params){};
	
	
	
	
	/* *******************************************************************************
	 * 其他方法
	 ******************************************************************************* */
    public void viewAccepted(View v) {
    	
        if(v instanceof MergeView){
        	if(logger.isInfoEnabled()){
        		logger.info(channel.getClusterName() + " -- MergeView["+v.size()+"]=" + v);
        	}
        }else{
        	if(logger.isInfoEnabled()){
        		logger.info(channel.getClusterName() + " -- View["+v.size()+"]=" + v);
        	}
    	}
        memberSize=v.size();
    }

    public void block() {
    	if(logger.isInfoEnabled()){
    		logger.info(channel.getClusterName() + " -- received BlockEvent");
    	}
    }

    public void unblock() {
    	if(logger.isInfoEnabled()){
    		logger.info(channel.getClusterName() + " -- received UnblockEvent");
    	}
    }


    public void channelConnected(Channel channel) {
    	if(logger.isInfoEnabled()){
    		logger.info(channel.getClusterName() + " -- channelConnected channel = " + channel);
    	}
    }

    public void channelDisconnected(Channel channel) {
    	if(logger.isInfoEnabled()){
    		logger.info(channel.getClusterName() + " -- channelDisconnected channel = " + channel);
    	}
    }

    public void channelClosed(Channel channel) {
    	if(logger.isInfoEnabled()){
    		logger.info(channel.getClusterName() + " -- channelClosed channel = " + channel);
    	}
    }

    public void channelShunned() {
    	if(logger.isInfoEnabled()){
    		logger.info(channel.getClusterName() + " -- channelShunned");
    	}
    }

    public void channelReconnected(Address addr) {
    	if(logger.isInfoEnabled()){
    		logger.info(channel.getClusterName() + " --channelReconnected Address = " + addr);
    	}
    }
    
    public void actionPerformed(ActionEvent e) {
    	if(logger.isInfoEnabled()){
    		logger.info(channel.getClusterName() + " -- actionPerformed e = " + e);
    	}
//        String     command=e.getActionCommand();
//        if("Clear".equals(command)) {
//            if(no_channel) {
//                clearPanel();
//                return;
//            }
//            sendClearPanelMsg();
//        }
//        else if("Leave".equals(command)) {
//            stop();
//        }
//        else
//            System.out.println("Unknown action");
    }
    
	public static void main(String[] args) {
		//测试
	}
	
	
}
